package www;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;

/**
 * hbase-demo
 *
 * reference:https://blog.csdn.net/vori2010/article/details/78536327
 */
public class App 
{

    static Configuration conf = null;
    static Connection conn = null;

    static {
        try {
            conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.property.clientPort", "2181");
            conf.set("hbase.zookeeper.quorum", "itcast01");
            conf.set("zookeeper.znode.parent", "/hbase");
            conn = ConnectionFactory.createConnection(conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static void main( String[] args)
    {

        try {
//            createTable("webtable", new String[]{"contents", "anchor", "people"});
            /*    --------    */
//            putData("com.cnn.www", "webtable", Bytes.toBytes("contents"), Bytes.toBytes("html"), Bytes.toBytes("webtable-com.cnn.www-contents:html=<html><head></head>...</html>"));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("contents"), Bytes.toBytes("html"), Bytes.toBytes("webtable-com.cnn.www-contents:html=<html><head></head>.1.</html>"));
//            putData("com.example.www", "webtable", Bytes.toBytes("contents"), Bytes.toBytes("html"), Bytes.toBytes("webtable-com.cnn.www-contents:html=<html><head></head>...</html>"));
//
//            putData("com.cnn.www", "webtable", Bytes.toBytes("anchor"), Bytes.toBytes("cnnsi.com"), Bytes.toBytes("CNN"));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("anchor"), Bytes.toBytes("my.look.ca"), Bytes.toBytes("CNN.com"));
//
//            putData("com.example.www", "webtable", Bytes.toBytes("people"), Bytes.toBytes("author"), Bytes.toBytes("John Doe"));
            /*    --------    */

//            listCells("webtable", "com.cnn.www");
////            /*    --------    */
//            listCells("webtable", "com.cnn.www", Bytes.toBytes("contents"), Bytes.toBytes("html"));
//            /*    --------    */
//            listCells("webtable", "com.cnn.www", Bytes.toBytes("contents"), Bytes.toBytes("html"));
//            /*    --------    */
//            modifyTableAddFamily("webtable", "visitflow", "Description", "往表中添加列族");
//            /*    --------    */
//            modifyColumn("webtable", "visitflow", Integer.MAX_VALUE);
//
//            /*    --------    */
//            putData("com.cnn.www", "webtable", Bytes.toBytes("visitflow"), Bytes.toBytes("pv"), Bytes.toBytes(10000L));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("visitflow"), Bytes.toBytes("pv"), Bytes.toBytes(20000L));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("visitflow"), Bytes.toBytes("pv"), Bytes.toBytes(30000L));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("visitflow"), Bytes.toBytes("pv"), Bytes.toBytes("测试变换值类型"));
//
//            /*    --------    */
//            listCells("webtable", "com.cnn.www", Bytes.toBytes("visitflow"), Bytes.toBytes("pv"));
//            /*    --------    */
//            addColumn("webtable", "contents", 3);
//
//            /*    --------    */
//            putData("com.cnn.www", "webtable", Bytes.toBytes("contents"), Bytes.toBytes("html"), Bytes.toBytes("webtable-com.cnn.www-contents:html=<html><head></head>...</html>"));
//            putData("com.cnn.www", "webtable", Bytes.toBytes("contents"), Bytes.toBytes("html"), Bytes.toBytes("webtable-com.cnn.www-contents:html=<html><head></head>.1.</html>"));
//
////            /*    --------    */
//            listCells("webtable", "com.cnn.www", Bytes.toBytes("contents"), Bytes.toBytes("html"));
//            /*    -----测试失败---    */
            deleteColumn("webtable", "com.cnn.www", Bytes.toBytes("contents"), Bytes.toBytes("html"));
        } catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println( "Hello World!" );
    }

    /**
     * 创建表
     *
     * @param tableName
     * @param family
     * @throws Exception
     */
    public static void createTable(String tableName, String[] family)
            throws Exception {
        if(StringUtils.isEmpty(tableName)) {
            return;
        }

        if(ArrayUtils.isEmpty(family)) {
            return;
        }

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();

        // 表 HTableDescriptor 列 HColumnDescriptor
        HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
        for (int i = 0; i < family.length; i++) {
            desc.addFamily(new HColumnDescriptor(family[i]));
        }
        if (admin.tableExists(tableName)) {
            System.out.println("table Exists!");
        } else {
            admin.createTable(desc);
            System.out.println("create table Success!");
        }
        admin.close();
    }

    /**
     * 插入(更新)数据
     *
     * @param rowKey
     * @param tableName
     * @param family
     * @param qualifier
     * @param value
     * @throws IOException
     */

    public static void putData(String rowKey, String tableName, byte[] family, byte[] qualifier, byte[] value) throws IOException {
        if (StringUtils.isEmpty(rowKey) || StringUtils.isEmpty(tableName)) {
            return;
        }

        if (ArrayUtils.isEmpty(family) || ArrayUtils.isEmpty(qualifier)) {
            return;
        }

        Put put = new Put(Bytes.toBytes(rowKey));
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // byte[] family, byte[] qualifier, byte[] value
        put.addColumn(family, qualifier, value);
        table.put(put);
        System.out.println("add data Success!");
        table.close();
    }

    /**
     * 遍历指定rowKey的数据
     *
     * @param tableName
     * @param rowKey
     * @return Result 可自定义某种数据封装
     * @throws IOException
     */
    public static Result listCells(String tableName, String rowKey) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        System.out.println("result.current(): " + result.current());

        CellScanner cellScanner = result.cellScanner();

        System.out.println("cellScanner: " + cellScanner.toString());
        System.out.println("cellScanner.current(): " + cellScanner.current());

        System.out.println("---------------------------------------------");

        for (Cell cell : result.listCells()) {
            System.out.println("cell: " + cell.toString());
            System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            System.out.println("value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            System.out.println("timestamp: " + cell.getTimestamp());
            System.out.println("---------------------------------------------");
        }

        return result;
    }

    /**
     * 遍历指定rowKey和列的数据
     *
     * @param tableName
     * @param rowKey
     * @return Result 可自定义某种数据封装
     * @throws IOException
     */
    public static Result listCells(String tableName, String rowKey, byte[] family, byte[] qualifier) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setMaxVersions(Integer.MAX_VALUE);
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        for (Cell cell : result.getColumnCells(family, qualifier)) {
            System.out.println("cell: " + cell.toString());
            System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            // value要根据实际类型转换toString等
            System.out.println("value: " + Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            System.out.println("timestamp: " + cell.getTimestamp());
            System.out.println("***************************************");
        }

        return result;
    }

    /**
     * 获取指定rowKey和列最新版本的数据
     *
     * @param tableName
     * @param rowKey
     * @return Result 可自定义某种数据封装
     * @throws IOException
     */
    public static Result latestCell(String tableName, String rowKey, byte[] family, byte[] qualifier) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        Cell cell = result.getColumnLatestCell(family, qualifier);
        System.out.println("cell: " + cell.toString());
        System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
        System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
        System.out.println("value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
        System.out.println("timestamp: " + cell.getTimestamp());

        return result;
    }

    /**
     * 修改表结构-替换该表的所有列，相当于重建表
     *
     * @param tableName
     * @param family
     * @param key
     * @param value
     * @throws Exception
     */
    public static void modifyTableAddFamily(String tableName, String family, String key, String value) throws Exception {
        if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(family) || StringUtils.isEmpty(key)) {
            return;
        }

        TableName tn = TableName.valueOf(tableName);

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        HTableDescriptor desc = new HTableDescriptor(tn);
        desc.addFamily(new HColumnDescriptor(family));
        desc.setValue(key, value);
        // 要先disable
        admin.disableTable(tn);
        admin.modifyTable(tn, desc);
        // 再enable
        admin.enableTable(tn);
        admin.close();
    }

    /**
     * 修改列属性[也可在创建时设置]
     *
     * @param tableName
     * @param family
     * @param maxVersions
     * @throws Exception
     */
    public static void modifyColumn(String tableName, String family, int maxVersions) throws Exception {
        if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(family)) {
            return;
        }

        TableName tn = TableName.valueOf(tableName);

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();

        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
        hColumnDescriptor.setMaxVersions(maxVersions);
        admin.disableTable(tn);
        admin.modifyColumn(tn, hColumnDescriptor);
        admin.enableTable(tn);
        admin.close();
    }

    /**
     * 添加列
     *
     * @param tableName
     * @param family
     * @param maxVersions
     * @throws Exception
     */
    public static void addColumn(String tableName, String family, int maxVersions) throws Exception {
        if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(family)) {
            return;
        }

        TableName tn = TableName.valueOf(tableName);

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();

        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
        hColumnDescriptor.setMaxVersions(maxVersions);
        admin.disableTable(tn);
        admin.addColumn(tn, hColumnDescriptor);
        admin.enableTable(tn);
        admin.close();
    }
    /**
     * 遍历查询表中全部数据
     *
     * @param tableName
     * @return ResultScanner
     * @throws IOException
     */
    public static ResultScanner listCells(String tableName) throws IOException {
        Scan scan = new Scan();
        // 返回版本数据
        scan.setMaxVersions(Integer.MAX_VALUE);

        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        ResultScanner resultScanner = table.getScanner(scan);

        Iterator iterator = resultScanner.iterator();

        while (iterator.hasNext()) {
            Result result = (Result) iterator.next();
            for (Cell cell : result.rawCells()) {
                System.out.println("cell: " + cell.toString());
                System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
                System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                System.out.println("value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                System.out.println("timestamp: " + cell.getTimestamp());
                System.out.println("***************************************");
            }
        }

        return resultScanner;
    }

    /**
     * 删除列
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @throws IOException
     */
    public static void deleteColumn(String tableName, String rowKey, byte[] family, byte[] qualifier) throws IOException {
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        // 删除指定列
        delete.addColumn(family, qualifier);
        table.delete(delete);
    }

    /**
     * 删除表
     *
     * @param tableName
     * @throws IOException
     */
    public static void deteleTable(String tableName) throws IOException {
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
        admin.close();
    }
}
